package mspbots.smileback.client.core.client;

import com.fasterxml.jackson.databind.JsonNode;
import lombok.extern.log4j.Log4j2;
import mspbots.smileback.client.annotation.ApiException;
import mspbots.smileback.client.oauth.AuthProperties;
import mspbots.smileback.client.oauth.AuthService;
import mspbots.smileback.common.QueryParams;
import mspbots.smileback.common.RabbitKeys;
import mspbots.smileback.common.ResultSync;
import org.springframework.amqp.rabbit.core.RabbitMessagingTemplate;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.util.UriComponentsBuilder;
import org.springframework.web.util.UriUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.URI;
import java.nio.charset.Charset;
import java.time.Duration;

/**
 * com.mspbots.api.client.WiseClient
 *
 * @author Alex bob(https://github.com/vnobo)
 * @date Created by 2019/11/12
 */
@Log4j2
@Service
public class SmileBackClient {

    private final WebClient webClient;

    private final AuthService authService;
    private final AuthProperties authProperties;

    private final RabbitMessagingTemplate rabbitMessagingTemplate;

    public SmileBackClient(WebClient.Builder builder,
                           AuthProperties authProperties,
                           AuthService authService,
                           RabbitMessagingTemplate rabbitMessagingTemplate) {
        this.webClient = builder.defaultHeader("Accept", "application/json").build();
        this.authService = authService;
        this.authProperties = authProperties;
        this.rabbitMessagingTemplate = rabbitMessagingTemplate;
    }

    public URI buildUrl(String path, QueryParams queryParams) {

        log.debug("Build path: {}, queryParams: {}", path, queryParams);

        return UriComponentsBuilder.fromHttpUrl(authProperties.getEndpoints())
                .path(path).queryParams(queryParams).build().toUri();
    }

    public Flux<JsonNode> fetch(URI url) {
        return this.request(url).expand(jsonNode -> {
            JsonNode nextLinkNode = jsonNode.get("next");
            if (ObjectUtils.isEmpty(nextLinkNode) || nextLinkNode.isNull()) {
                return Mono.empty();
            }
            String nextUrl = nextLinkNode.textValue();
            nextUrl = UriUtils.decode(nextUrl, Charset.defaultCharset());
            return this.request(UriComponentsBuilder.fromUriString(nextUrl).build().toUri());
        });
    }

    private Mono<JsonNode> request(URI uri) {
        log.debug("Request uri: {}", uri.toString());
        return Mono.subscriberContext().flatMap(ctx -> this.authService.getToken(ctx.get("tenantId"))
                .flatMap(token -> this.webClient.get()
                        .uri(uri).headers(httpHeaders -> httpHeaders.setBearerAuth(token))
                        .retrieve()
                        .onStatus(HttpStatus::isError, clientResponse -> clientResponse.bodyToMono(JsonNode.class)
                                .map(errMsg -> new ApiException(errMsg, clientResponse.statusCode())))
                        .bodyToMono(JsonNode.class))
                .timeout(Duration.ofMinutes(5))
                .doOnSuccess(resp -> this.handlerSuccessNext(resp, ctx.get("resultSync")))
                .onErrorResume(err -> this.handlerException(err, ctx.get("resultSync"))));
    }

    private Mono<JsonNode> handlerException(Throwable err, ResultSync resultSync) {
        if (err instanceof ApiException) {
            resultSync.setMsg(((ApiException) err).getMsg());
            resultSync.setStatus(((ApiException) err).getStatus());
        } else {
            resultSync.setMsg(err.getMessage());
            resultSync.setStatus(HttpStatus.INTERNAL_SERVER_ERROR);
        }

        resultSync.setElapsedTime(System.currentTimeMillis() - resultSync.getElapsedTime());

        log.error("Sync {} data error action {}, elapsedTime {}, msg: {}",
                resultSync.getTenantId(), resultSync.getAction(),
                resultSync.getElapsedTime(), resultSync.getMsg());

        this.rabbitMessagingTemplate.convertAndSend(RabbitKeys.BASE_EXCHANGE_NAME,
                RabbitKeys.LOGS_TASK_QUEUE_KEY, resultSync);
        return Mono.error(err);
    }

    private void handlerSuccessNext(JsonNode resp, ResultSync resultSync) {
        JsonNode countNode = resp.get("count");

        if (ObjectUtils.isEmpty(countNode)) {
            resultSync.setCount(resp.withArray("results").size());
        } else {
            resultSync.setCount(countNode.intValue());
        }

        resultSync.setElapsedTime(System.currentTimeMillis() - resultSync.getElapsedTime());

        log.info("sync {} data success, action {}, elapsedTime {}, count {}",
                resultSync.getTenantId(), resultSync.getAction(),
                resultSync.getElapsedTime(), resultSync.getCount());

        this.rabbitMessagingTemplate.convertAndSend(RabbitKeys.BASE_EXCHANGE_NAME,
                RabbitKeys.LOGS_TASK_QUEUE_KEY, resultSync);
    }
}
